package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TradeTrademarkCategoryUserRefundBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import com.bw.gmall.realtime.common.util.HbaseUtil;
import com.bw.gmall.realtime.dws.function.DorisMapFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.time.Duration;
import java.util.HashSet;

/*
12.11交易域品牌-品类-用户粒度退单各窗口汇总表（练习）
 */

public class DwsTradeTrademarkCategoryUserRefundWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeTrademarkCategoryUserRefundWindow().start(Constant.TOPIC_DWD_TRADE_ORDER_REFUND,
                Constant.DWS_TRADE_TRADEMARK_CATEGORY_USER_REFUND_WINDOW,4,10032);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        // 1.ETL数据清理
        SingleOutputStreamOperator<JSONObject> etlStream = etl(dataStreamSource);

        // 2.设置水位线，映射实体类
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> processStream = getProcessStream(etlStream);

        // 3.关联维度
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> mapStream = getMapStream(processStream);

        // 4.分组开窗聚合
        SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> resultStream = getResultStream(mapStream);
        resultStream.print();

        // 5.写到doris
        resultStream.map(new DorisMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink(Constant.DWS_TRADE_TRADEMARK_CATEGORY_USER_REFUND_WINDOW));

    }

    private static SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getResultStream(SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> mapStream) {
        return mapStream.keyBy(new KeySelector<TradeTrademarkCategoryUserRefundBean, Tuple5<String, String, String, String, String>>() {
                    @Override
                    public Tuple5<String, String, String, String, String> getKey(TradeTrademarkCategoryUserRefundBean t) throws Exception {
                        return new Tuple5<>(t.getTrademarkId(), t.getCategory1Id(), t.getCategory2Id(), t.getCategory3Id(), t.getUserId());
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeTrademarkCategoryUserRefundBean>() {
                    @Override
                    public TradeTrademarkCategoryUserRefundBean reduce(TradeTrademarkCategoryUserRefundBean x, TradeTrademarkCategoryUserRefundBean y) throws Exception {
                        x.getOrderIdSet().addAll(y.getOrderIdSet());
                        return x;
                    }
                }, new ProcessWindowFunction<TradeTrademarkCategoryUserRefundBean, TradeTrademarkCategoryUserRefundBean, Tuple5<String, String, String, String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple5<String, String, String, String, String> stringStringStringStringStringTuple5, ProcessWindowFunction<TradeTrademarkCategoryUserRefundBean, TradeTrademarkCategoryUserRefundBean, Tuple5<String, String, String, String, String>, TimeWindow>.Context context, Iterable<TradeTrademarkCategoryUserRefundBean> iterable, Collector<TradeTrademarkCategoryUserRefundBean> collector) throws Exception {
                        TradeTrademarkCategoryUserRefundBean tradeTrademarkCategoryUserRefundBean = iterable.iterator().next();
                        tradeTrademarkCategoryUserRefundBean.setStt(DateFormatUtil.tsToDateTime(context.window().getStart()));
                        tradeTrademarkCategoryUserRefundBean.setEdt(DateFormatUtil.tsToDateTime(context.window().getEnd()));
                        tradeTrademarkCategoryUserRefundBean.setCurDate(DateFormatUtil.tsToDate(System.currentTimeMillis()));
                        tradeTrademarkCategoryUserRefundBean.setRefundCount((long) tradeTrademarkCategoryUserRefundBean.getOrderIdSet().size());
                        collector.collect(tradeTrademarkCategoryUserRefundBean);
                    }
                });
    }

    private static SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getMapStream(SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> processStream){
        return processStream.map(new RichMapFunction<TradeTrademarkCategoryUserRefundBean, TradeTrademarkCategoryUserRefundBean>() {
            Connection hbaseConnect;

            @Override
            public void open(Configuration parameters) throws Exception {
                hbaseConnect = HbaseUtil.getHbaseConnect();
            }

            @Override
            public void close() throws Exception {
                hbaseConnect.close();
            }

            @Override
            public TradeTrademarkCategoryUserRefundBean map(TradeTrademarkCategoryUserRefundBean tradeTrademarkCategoryUserRefundBean) throws Exception {
                //sku
                JSONObject dimSkuInfo = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_sku_info", tradeTrademarkCategoryUserRefundBean.getSkuId());
                String tmId = dimSkuInfo.getString("tm_id");
                tradeTrademarkCategoryUserRefundBean.setTrademarkId(tmId);
                String category3Id = dimSkuInfo.getString("category3_id");

                //tm
                JSONObject dimBaseTrademark = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_base_trademark", tmId);
                tradeTrademarkCategoryUserRefundBean.setTrademarkName(dimBaseTrademark.getString("tm_name"));

                //c3
                JSONObject dimBaseCategory3 = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_base_category3", category3Id);
                tradeTrademarkCategoryUserRefundBean.setCategory3Id(category3Id);
                tradeTrademarkCategoryUserRefundBean.setCategory3Name(dimBaseCategory3.getString("name"));
                String category2Id = dimBaseCategory3.getString("category2_id");

                //c2
                JSONObject dimBaseCategory2 = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_base_category2", category2Id);
                tradeTrademarkCategoryUserRefundBean.setCategory2Id(category2Id);
                tradeTrademarkCategoryUserRefundBean.setCategory2Name(dimBaseCategory2.getString("name"));
                String category1Id = dimBaseCategory2.getString("category1_id");

                //c1
                JSONObject dimBaseCategory1 = HbaseUtil.getCells(hbaseConnect, Constant.HBASE_NAMESPACE, "dim_base_category1", category1Id);
                tradeTrademarkCategoryUserRefundBean.setCategory1Id(category1Id);
                tradeTrademarkCategoryUserRefundBean.setCategory1Name(dimBaseCategory1.getString("name"));

                return tradeTrademarkCategoryUserRefundBean;
            }
        });
    }

    private static SingleOutputStreamOperator<TradeTrademarkCategoryUserRefundBean> getProcessStream(SingleOutputStreamOperator<JSONObject> etlStream) {
        return etlStream.assignTimestampsAndWatermarks(WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, timestamp) -> event.getLong("ts"))
                        .withIdleness(Duration.ofSeconds(120)))
                .keyBy(x -> x.getString("id"))
                .process(new KeyedProcessFunction<String, JSONObject, TradeTrademarkCategoryUserRefundBean>() {
                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TradeTrademarkCategoryUserRefundBean>.Context context, Collector<TradeTrademarkCategoryUserRefundBean> collector) throws Exception {
                        //创建hashSet存储orderid，用于统计退单次数
                        HashSet<String> set = new HashSet<>();
                        set.add(jsonObject.getString("order_id"));
                        //赋值实体类
                        collector.collect(TradeTrademarkCategoryUserRefundBean.builder()
                                .userId(jsonObject.getString("user_id"))
                                .ts(jsonObject.getLong("ts"))
                                .skuId(jsonObject.getString("sku_id"))
                                .orderIdSet(set)
                                .build());
                    }
                });
    }

    private static SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                //回撤流
                if (s != null) {
                    try {
                        JSONObject jsonObject = JSON.parseObject(s);
                        String userId = jsonObject.getString("user_id");
                        String skuId = jsonObject.getString("sku_id");
                        Long ts = jsonObject.getLong("ts");
                        if (userId != null && skuId != null && ts != null) {
                            jsonObject.put("ts", ts * 1000);
                            collector.collect(jsonObject);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });
    }
}
